package czk.concrent.threadpool.ownthreadpool;

import java.io.IOException;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 自定义线程池的练习
 *
 * @Author:ChenZhangKun
 * @Date: 2021/1/27 19:40
 */
public class MyThreadPoolPractice {
    public static void main(String[] args) throws IOException {
        MyThreadPool threadPool = new MyThreadPool(5, 1L, TimeUnit.SECONDS, 5, 10);
        for (int i = 0; i < 10; i++) {
            int j = i;
            threadPool.execute(() -> System.out.println(Thread.currentThread().getName() + "  " + j));
        }
    }
}

/**
 * 线程池
 */
class MyThreadPool {
    /**
     * 核心线程数
     */
    private int coreNum;
    /**
     * 超时时间
     */
    private Long time;
    /**
     * 时间单位
     */
    private TimeUnit timeUnit;
    /**
     * 阻塞队列
     */
    private MyBlockingQueue<Runnable> blockingQueue;
    /**
     * 存放线程的集合
     */
    Set<MyWork> set = new HashSet<>(10);
    private Integer maxThreadNum;

    public MyThreadPool(int coreNum, Long time, TimeUnit timeUnit, int capacity, int maxThreadNum) {
        this.coreNum = coreNum;
        this.time = time;
        this.timeUnit = timeUnit;
        blockingQueue = new MyBlockingQueue(capacity);
        this.maxThreadNum = maxThreadNum;
    }

    public void execute(Runnable task) {
        synchronized (set) {
            // 判断核心线程数
            if (set.size() < coreNum) {
                // 新建线程去执行
                MyWork myWork = new MyWork(task);
                // 加入线程集合去处理
                set.add(myWork);
                //
                System.out.println("加入线程集合啦");
                myWork.start();
            } else {
                // 加入阻塞队列
                System.out.println("加入阻塞队列了");
                this.blockingQueue.offer(task, time, TimeUnit.SECONDS);
                if (this.blockingQueue.isFull()) {
                    System.out.println("开始启用最大线程");
                    if (set.size() < maxThreadNum) {
                        // 加到最大线程数
                        MyWork myWork = new MyWork(task);
                        set.add(myWork);
                        myWork.start();
                    } else {
                        // 执行拒绝策略
                        System.out.println("线程池和任务队列已满，不能新增任务");
                    }

                }
            }
        }
    }

    class MyWork extends Thread {
        private Runnable runnable;

        public MyWork(Runnable target) {
            this.runnable = target;
        }

        @Override
        public void run() {
            // 执行任务
            while (runnable != null || (runnable = blockingQueue.take(1, TimeUnit.SECONDS)) != null) {
                // 执行任务
                try {
                    runnable.run();
                    System.out.println("正在执行任务");
                } finally {
                    runnable = null;
                }

            }
            // 娶不到任务时移除当前线程
            synchronized (set) {
                while (set.size() > coreNum) {
                    System.out.println("移除当前线程" + this);
                    set.remove(this);
                }

            }
        }
    }

}

/**
 * 自己的阻塞队列
 *
 * @param <T>
 */
class MyBlockingQueue<T> {
    /**
     * 大小
     */
    private int capacity;
    /**
     * 双向链表
     */
    private Deque<T> deque = new ArrayDeque<>(16);
    /**
     * 锁
     */
    private ReentrantLock lock = new ReentrantLock();
    /**
     * 休息室1
     */
    private Condition fullCondition = lock.newCondition();
    /**
     * 休息室2
     */
    private Condition emptyCondition = lock.newCondition();

    public MyBlockingQueue(int capacity) {
        this.capacity = capacity;
    }

    /**
     * 带超时时间的添加
     *
     * @param t        任务
     * @param minute   时间
     * @param timeUnit 时间单位
     * @return
     */
    public boolean offer(T t, long minute, TimeUnit timeUnit) {
        try {
            // 上锁
            lock.lock();
            // 转为纳秒
            long nanos = timeUnit.toNanos(minute);
            // 当队列满时候
            while (isFull()) {
                // 满了等待nanos秒
                if (nanos < 0) {
                    // 抛出异常
                    return false;
                }
                try {
                    // 带超时的等待
                    nanos = fullCondition.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            // 添加成功
            deque.add(t);
            System.out.println("任务加入队列成功");
            // 唤醒消费休息室
            emptyCondition.signal();
        } finally {
            // 解锁
            lock.unlock();
        }
        return true;
    }

    /**
     * 超时获取
     *
     * @param minute
     * @param timeUnit
     * @return
     */
    public T take(long minute, TimeUnit timeUnit) {
        T first;
        try {
            // 阻塞
            lock.lock();
            // 转为纳秒
            long nanos = timeUnit.toNanos(minute);
            while (isEmpty()) {
                // 超时直接返回null
                if (nanos < 0) {
                    System.out.println("在规定时间内没有获取到");
                    return null;
                }
                // 等待
                try {
                    nanos = emptyCondition.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            // 获取到
            first = deque.removeFirst();
            // 或许其他线程去放任务
            fullCondition.signal();
        } finally {
            // 释放
            lock.unlock();
        }
        return first;
    }

    /**
     * 判断队列是否为空
     *
     * @return
     */
    private boolean isEmpty() {
        return deque.size() == 0;
    }

    /**
     * 判断队列是否满了
     *
     * @return
     */
    public boolean isFull() {
        return deque.size() == capacity;
    }
}
